from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType
import pandas as pd
from pyspark.sql import functions as F
import pymysql

# 3查询高分电影中打分次数最多的用户, 此人打分的平均分,最高分，最低分(wan)
if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("xixi3").\
        getOrCreate()
    sc = spark.sparkContext
    # 1. 读取数据集
    schema = StructType().add("user_id", StringType(), nullable=True). \
        add("movie_id", IntegerType(), nullable=True). \
        add("rank", IntegerType(), nullable=True). \
        add("ts", StringType(), nullable=True)
    df = spark.read.format("csv"). \
        option("sep", "::"). \
        option("header", False). \
        option("encoding", "utf-8"). \
        schema(schema=schema). \
        load("hdfs://node1:8020/input/movie.csv")
    # 找出这个人
    user_id = df.where("rank > 3"). \
        groupBy("user_id"). \
        count(). \
        withColumnRenamed("count", "cnt"). \
        orderBy("cnt", ascending=False). \
        limit(1). \
        first()['user_id']
    # 计算这个人的打分平均分
    df.filter(df['user_id'] == user_id). \
        select(F.round(F.avg("rank"), 2),F.round(F.max("rank"), 2),F.round(F.min("rank"), 2)). \
        withColumnRenamed("round(avg(rank), 2)", "avg"). \
        withColumnRenamed("round(max(rank), 2)", "max"). \
        withColumnRenamed("round(min(rank), 2)", "min"). \
        write.mode("overwrite"). \
        format("jdbc"). \
        option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true"). \
        option("dbtable", "find_max_amm"). \
        option("user", "root"). \
        option("password", "123456"). \
        save()
